package com.imooc.bilibili.config;

import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.constant.rocket.UserMomentsConstant;
import com.imooc.bilibili.domain.following.UserFollowing;
import com.imooc.bilibili.domain.user.UserMoment;
import com.imooc.bilibili.following.service.UserFollowingService;
import com.imooc.bilibili.util.StringUtils;
import com.imooc.bilibili.util.json.JsonUtils;
import com.imooc.bilibili.util.redis.RedisUtils;
import com.imooc.bilibili.websocket.WebSocketService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.ArrayList;
import java.util.List;

/**
 * RocketMQ 配置
 *
 * @author gcq
 * @Create 2022-05-28
 */
//@Configuration
public class RocketMQConfig {

    private final Logger logger = LoggerFactory.getLogger(RocketMQConfig.class);

    @Value("${rocketmq.name.server.address}")
    private String nameServerAddress;

    @Autowired
    private RedisUtils redisUtils;

    @Autowired
    private UserFollowingService userFollowingService;

    /**
     * 用户动态生产者
     *
     * @return
     */
    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws MQClientException {
        DefaultMQProducer mqProducer = new DefaultMQProducer(UserMomentsConstant.GROUP_MOMENTS);
        mqProducer.setNamesrvAddr(nameServerAddress);
        mqProducer.start();
        return mqProducer;
    }

    /**
     * 用户动他消费者
     *
     * 消费用户动态呢容
     * 相关业务场景：
     * 电商：
     *      购买后，库存扣减等等
     *
     * @return
     * @throws MQClientException
     */
    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentConsumer() throws MQClientException {
        System.out.println("开始消费消息");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_MOMENTS);
        consumer.setNamesrvAddr(nameServerAddress);
        // 订阅生产者 二级主题 主题所有的内容
        consumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, "*");
        // 监听器，处理内容
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            /**
             * 处理MQ信息
             *
             * @param messageExtList 消息扩充
             * @param consumeConcurrentlyContext  消费内容上下文
             * @return 状态
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt message = messageExtList.get(0);
                if(message == null) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String bodyStr = new String(message.getBody());
                UserMoment userMoment = JsonUtils.jsonToPojo(bodyStr, UserMoment.class);
                Long userId = userMoment.getUserId();

                // 获取用户粉丝，得到哪些用户订阅了当前的用户
                // 更新用户动态内容
                // 把用户发送的内容放到redis中，通过reids去查询
                List<UserFollowing> userFanList = userFollowingService.getUserFans(userId);
                for (UserFollowing fan : userFanList) {
                    String key = "subscribed-" + fan.getUserId();
                    String subscribedListStr = redisUtils.get(key);
                    List<UserMoment> subscribedList;
                    if(StringUtils.isEmpty(subscribedListStr)) {
                        subscribedList = new ArrayList<>();
                    } else {
                        subscribedList = JsonUtils.jsonToList(subscribedListStr, UserMoment.class);
                    }
                    // 添加当前动态
                    subscribedList.add(userMoment);
                    redisUtils.set(key, JsonUtils.objectToJson(subscribedList));
                }
                for (MessageExt messageExt : messageExtList) {
                    System.out.println("消息" + messageExt);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }

    /**
     * 弹幕生产者
     *
     * @return
     */
    @Bean("danmusProducer")
    public DefaultMQProducer danmusProducer() throws MQClientException {
        DefaultMQProducer mqProducer = new DefaultMQProducer(UserMomentsConstant.GROUP_DANMUS);
        mqProducer.setNamesrvAddr(nameServerAddress);
        mqProducer.start();
        return mqProducer;
    }

    /**
     * 弹幕消费者
     * 获取到sessionId和Message
     * 根据sessionId获取到当前WebSocket链接，然后发送消息
     *
     * @return
     * @throws MQClientException
     */
    @Bean("danmusConsumer")
    public DefaultMQPushConsumer danmusConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_DANMUS);
        consumer.setNamesrvAddr(nameServerAddress);
        // 订阅生产者 二级主题 主题所有的内容
        consumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, "*");
        // 监听器，处理内容
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt msg = messageExtList.get(0);
                byte[] msgByte = msg.getBody();
                String bodyStr = new String(msgByte);
                JSONObject jsonObject = JsonUtils.jsonToPojo(bodyStr, JSONObject.class);
                String sessionId = jsonObject.getString("sessionId");
                String message = jsonObject.getString("message");
                WebSocketService webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);
                if(webSocketService.getSession().isOpen()){
                    try {
                        webSocketService.sendMessage(message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                // 标记该消息已经被消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }
}